package com.ywu.mq.standard.service;

import com.ywu.app.manage.spi.dto.YwuAppDto;
import com.ywu.app.manage.spi.spi.YwuAppSpi;
import com.ywu.boot.context.YwuContext;
import com.ywu.boot.context.YwuContextHolder;
import com.ywu.common.entity.PageList;
import com.ywu.common.page.PageService;
import com.ywu.common.result.CommonResult;
import com.ywu.common.utils.TimerIdGenerateUtil;
import com.ywu.mq.sdk.domain.dto.YwuKafkaDto;
import com.ywu.mq.sdk.domain.entity.YwuKafkaConsumerLogEntity;
import com.ywu.mq.sdk.domain.entity.YwuKafkaProduceMessageEntity;
import com.ywu.mq.standard.data.YwuKafkaRepository;
import com.ywu.mq.standard.data.YwuKafkaTopicRepository;
import com.ywu.mq.standard.entity.YwuKafkaTopicEntity;
import com.ywu.mq.standard.entity.YwuKafkaTopicSubEntity;
import com.ywu.mq.standard.produce.YwuKafkaProduce;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

/**
 * @ClassName YwuKafkaService
 * @Description TODO
 * @Author GroundDemo
 * @Date 2025/3/28 10:09
 * @Version 1.0
 **/
@Service
@Slf4j
public class YwuKafkaService {

    @Autowired
    private YwuKafkaTopicRepository kafkaTopicRepository;

    @Autowired
    private YwuKafkaRepository repository;

    @Autowired
    private YwuKafkaProduce produce;

    @Autowired
    private YwuAppSpi ywuAppSpi;

    private static final String YWU_MQ_APP_ID = "com.ywu.mq.root";

    /**
    * @Author GroundDemo
    * @Description 生产消息
    * @Date 10:14 2025/3/28
    * @Param [ywuKafkaDto]
    * @return com.ywu.common.result.CommonResult<java.lang.Object>
    **/
    public CommonResult<Object> produceInfo(YwuKafkaDto ywuKafkaDto) {
        // 查询主题是否存在
        YwuKafkaTopicEntity search = new YwuKafkaTopicEntity();
        search.setId(ywuKafkaDto.getTopicId());
        YwuKafkaTopicEntity queryEntity = kafkaTopicRepository.getEntity(search);
        if (Objects.isNull(queryEntity)) {
            return CommonResult.error().setMessage("topic is not exist");
        }
        // 构建主题信息
        produce.setTopic(queryEntity.getTopicName());
        produce.setKey(ywuKafkaDto.getKey());
        produce.setAppId(YWU_MQ_APP_ID);
        produce.setTopicAppId(queryEntity.getAppId());
        // 发送消息
        produce.produce(ywuKafkaDto.getValue());
        return CommonResult.success(null);
    }

    /**
     * @Author GroundDemo
     * @Description 重发消息
     * @Date 20:01 2025/3/28
     * @Param [ywuKafkaDto]
     * @return com.ywu.common.result.CommonResult<java.lang.Object>
     **/
    public CommonResult<Object> produceResendInfo(YwuKafkaDto ywuKafkaDto) {
        // 根据ID查询消息
        YwuKafkaProduceMessageEntity messageEntity = repository.getMessageEntity(ywuKafkaDto);
        if (Objects.isNull(messageEntity)) {
            return CommonResult.error().setMessage("message is not exist");
        }
        // 构建主题信息
        produce.setTopic(messageEntity.getTopicName());
        produce.setKey(messageEntity.getKey());
        produce.setAppId(YWU_MQ_APP_ID);
        produce.setTopicAppId(messageEntity.getTopicAppId());
        // 发送消息
        produce.produce(messageEntity.getValue());
        return CommonResult.success(null);
    }

    /**
    * @Author GroundDemo
    * @Description 记录发送日志
    * @Date 10:37 2025/3/28
    * @Param [messageEntity]
    * @return com.ywu.common.result.CommonResult<java.lang.Object>
    **/
    public CommonResult<Object> produceCreateMessage(YwuKafkaProduceMessageEntity messageEntity) {
        messageEntity.createInfo("system");
        repository.addKafkaProduceMessage(messageEntity);
        return CommonResult.success(null);
    }

    /**
    * @Author GroundDemo
    * @Description 校验主题是否存在
    * @Date 11:02 2025/3/28
    * @Param [messageEntity]
    * @return com.ywu.common.result.CommonResult<java.lang.Object>
    **/
    public CommonResult<Object> produceCheckTopic(YwuKafkaProduceMessageEntity messageEntity) {
        // 校验应用秘钥
        YwuAppDto ywuAppDto = new YwuAppDto();
        ywuAppDto.setAppToken(messageEntity.getAppKey());
        ywuAppDto.setAppId(messageEntity.getAppId());
        CommonResult<Object> result = ywuAppSpi.checkAppToken(ywuAppDto);
        if (Objects.isNull(result) || !result.getCode().equals("200")) {
            log.debug("app key check error");
            return CommonResult.error().setErrorCode("app.key.error").setMessage("app key check error");
        }
        // 如果是MQ平台应用，直接返回成功即可
        if (messageEntity.getAppId().equals(YWU_MQ_APP_ID)) {
            return CommonResult.success(null);
        }
        // 校验权限
        YwuKafkaDto ywuKafkaDto = new YwuKafkaDto();
        ywuKafkaDto.setAppId(messageEntity.getAppId());
        ywuKafkaDto.setTopicName(messageEntity.getTopicName());
        ywuKafkaDto.setTopicAppId(messageEntity.getTopicAppId());
        YwuKafkaTopicEntity ywuKafkaTopicEntity = kafkaTopicRepository.queryEntityByAppIdPermission(ywuKafkaDto);
        if (Objects.isNull(ywuKafkaTopicEntity)) {
            return CommonResult.error().setMessage("topic is not exist OR topic permission is refused");
        } else {
            return CommonResult.success(null);
        }
    }

    /**
    * @Author GroundDemo
    * @Description 查询主题的消息列表
    * @Date 12:34 2025/3/28
    * @Param [ywuKafkaDto]
    * @return com.ywu.common.result.CommonResult<java.lang.Object>
    **/
    public CommonResult<Object> queryMessageInfoList(YwuKafkaDto ywuKafkaDto) {
        // 构建分页参数
        PageService.execPageStartAndEnd(ywuKafkaDto.getPageInfo());
        PageList<YwuKafkaProduceMessageEntity> pageListRes = new PageList<>();
        pageListRes.setPageIndex(ywuKafkaDto.getPageInfo().getPageIndex());
        pageListRes.setPageSize(ywuKafkaDto.getPageInfo().getPageSize());
        // 如果appId没有传，返回空结果
        if (StringUtils.isEmpty(ywuKafkaDto.getTopicAppId()) || StringUtils.isEmpty(ywuKafkaDto.getTopicName())) {
            pageListRes.setDataList(new ArrayList<>());
            pageListRes.setRecordCount(0);
            return CommonResult.success(pageListRes);
        }
        // 查询数量
        int count = repository.queryMessageCount(ywuKafkaDto);
        if (count == 0) {
            pageListRes.setDataList(new ArrayList<>());
        } else {
            // 查询kafka消息列表
            List<YwuKafkaProduceMessageEntity> entities = repository.queryMessageList(ywuKafkaDto);
            if (CollectionUtils.isEmpty(entities)) {
                pageListRes.setDataList(new ArrayList<>());
            } else {
                pageListRes.setDataList(entities);
            }
        }
        pageListRes.setRecordCount(count);
        return CommonResult.success(pageListRes);
    }

    /**
    * @Author GroundDemo
    * @Description 主题与应用订阅授权
    * @Date 14:29 2025/3/28
    * @Param [ywuKafkaDto]
    * @return com.ywu.common.result.CommonResult<java.lang.Object>
    **/
    public CommonResult<Object> topicAppSub(YwuKafkaDto ywuKafkaDto) {
        YwuContext currentContext = YwuContextHolder.getCurrentContext();
        // 校验授权关系是否已经存在
        YwuKafkaTopicSubEntity search = new YwuKafkaTopicSubEntity();
        search.setAppId(ywuKafkaDto.getAppId());
        search.setTopicId(ywuKafkaDto.getTopicId());
        YwuKafkaTopicSubEntity queryEntity = repository.getTopicSubEntity(search);
        if (Objects.nonNull(queryEntity)) {
            // 授权关系已存在
            return CommonResult.error().setMessage("topic app sub has exist");
        }
        // 插入订阅信息
        YwuKafkaTopicSubEntity insertEntity = new YwuKafkaTopicSubEntity();
        insertEntity.setId(TimerIdGenerateUtil.nextId());
        insertEntity.setAppId(ywuKafkaDto.getAppId());
        insertEntity.setTopicId(ywuKafkaDto.getTopicId());
        insertEntity.createInfo(currentContext.getYwuContextUser().getUserAccount());
        repository.insertTopicAppSub(insertEntity);
        return CommonResult.success(null);
    }

    /**
    * @Author GroundDemo
    * @Description 主题与订阅授权删除
    * @Date 14:31 2025/3/28
    * @Param [ywuKafkaDto]
    * @return com.ywu.common.result.CommonResult<java.lang.Object>
    **/
    public CommonResult<Object> topicSubDeleteByAppId(YwuKafkaDto ywuKafkaDto) {
        // 根据appId删除授权信息
        repository.deleteTopicSubByAppId(ywuKafkaDto);
        return CommonResult.success(null);
    }

    /**
    * @Author GroundDemo
    * @Description 查询订阅列表
    * @Date 14:50 2025/3/28
    * @Param [ywuKafkaDto]
    * @return com.ywu.common.result.CommonResult<java.lang.Object>
    **/
    public CommonResult<Object> topicAppSubQuery(YwuKafkaDto ywuKafkaDto) {
        PageList<YwuKafkaTopicSubEntity> pageListRes = new PageList<>();
        pageListRes.setPageIndex(ywuKafkaDto.getPageInfo().getPageIndex());
        pageListRes.setPageSize(ywuKafkaDto.getPageInfo().getPageSize());
        // 构建分页参数
        PageService.execPageStartAndEnd(ywuKafkaDto.getPageInfo());
        // 如果appId没有传，返回空结果
        if (StringUtils.isEmpty(ywuKafkaDto.getTopicId())) {
            pageListRes.setDataList(new ArrayList<>());
            pageListRes.setRecordCount(0);
            return CommonResult.success(pageListRes);
        }
        // 查询数量
        int count = repository.queryTopicSubCount(ywuKafkaDto);
        if (count == 0) {
            pageListRes.setDataList(new ArrayList<>());
        } else {
            // 查询kafka主题订阅列表
            List<YwuKafkaTopicSubEntity> entities = repository.queryTopicSubList(ywuKafkaDto);
            if (CollectionUtils.isEmpty(entities)) {
                pageListRes.setDataList(new ArrayList<>());
            } else {
                pageListRes.setDataList(entities);
            }
        }
        pageListRes.setRecordCount(count);
        return CommonResult.success(pageListRes);
    }

    /**
    * @Author GroundDemo
    * @Description 根据ID删除授权
    * @Date 15:00 2025/3/28
    * @Param [ywuKafkaDto]
    * @return com.ywu.common.result.CommonResult<java.lang.Object>
    **/
    public CommonResult<Object> deleteTopicSubById(YwuKafkaDto ywuKafkaDto) {
        YwuKafkaTopicSubEntity deleteEntity = new YwuKafkaTopicSubEntity();
        deleteEntity.setId(ywuKafkaDto.getId());
        repository.deleteTopicSubById(deleteEntity);
        return CommonResult.success(null);
    }

    /**
    * @Author GroundDemo
    * @Description 校验当前主题是否有消费的权限
    * @Date 16:49 2025/3/28
    * @Param [ywuKafkaDto]
    * @return com.ywu.common.result.CommonResult<java.lang.Object>
    **/
    public CommonResult<Object> checkConsumerPermission(YwuKafkaDto ywuKafkaDto) {
        YwuKafkaProduceMessageEntity ywuKafkaProduceMessageEntity = new YwuKafkaProduceMessageEntity();
        ywuKafkaProduceMessageEntity.setAppKey(ywuKafkaDto.getAppKey());
        ywuKafkaProduceMessageEntity.setAppId(ywuKafkaDto.getAppId());
        ywuKafkaProduceMessageEntity.setTopicName(ywuKafkaDto.getTopicName());
        ywuKafkaProduceMessageEntity.setTopicAppId(ywuKafkaDto.getTopicAppId());
        return produceCheckTopic(ywuKafkaProduceMessageEntity);
    }

    /**
    * @Author GroundDemo
    * @Description 记录消费日志
    * @Date 18:55 2025/3/28
    * @Param [ywuKafkaConsumer]
    * @return com.ywu.common.result.CommonResult<java.lang.Object>
    **/
    public CommonResult<Object> createConsumerLog(YwuKafkaConsumerLogEntity ywuKafkaConsumer) {
        ywuKafkaConsumer.setTopicName("");
        ywuKafkaConsumer.createInfo("system");
        // 插入消费日志
        repository.insertConsumerLog(ywuKafkaConsumer);
        return CommonResult.success(null);
    }

    /**
    * @Author GroundDemo
    * @Description 查询kafka消费记录
    * @Date 19:49 2025/3/28
    * @Param [ywuKafkaConsumer]
    * @return com.ywu.common.result.CommonResult<java.lang.Object>
    **/
    public CommonResult<Object> queryKafkaConsumerLog(YwuKafkaConsumerLogEntity ywuKafkaConsumer) {
        // 根据消息ID查询消息记录
        YwuKafkaDto searchDto = new YwuKafkaDto();
        searchDto.setId(ywuKafkaConsumer.getMessageId());
        YwuKafkaProduceMessageEntity ywuKafkaProduceMessageEntity = repository.getMessageEntity(searchDto);
        // 根据消息ID查询消费记录
        List<YwuKafkaConsumerLogEntity> resList = repository.queryConsumerListByMessageId(ywuKafkaConsumer);
        resList.forEach(res -> {
            res.setTopicName(ywuKafkaProduceMessageEntity.getTopicName());
        });
        return CommonResult.success(resList);
    }
}
